//! 命名服务实现

use async_trait::async_trait;
use nacos_core::{
    NamingService, ServiceInstance, Result, ServiceStorage, EventListener, EventPublisher, NacosError, NetworkService, Event,
};
use std::sync::Arc;
use std::fmt;
use tokio::sync::RwLock;
use tracing::{info, instrument, warn};

use crate::model::{ServiceRegistry, ServiceStatus};

pub struct NamingServiceImpl {
    service_storage: Arc<dyn ServiceStorage>,
    event_publisher: Arc<dyn EventPublisher>,
    network_service: Arc<dyn NetworkService>,
    service_registry: Arc<RwLock<ServiceRegistry>>,
}

impl fmt::Debug for NamingServiceImpl {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("NamingServiceImpl")
         .field("service_storage", &"Arc<dyn ServiceStorage>")
         .field("event_publisher", &"Arc<dyn EventPublisher>")
         .field("network_service", &"Arc<dyn NetworkService>")
         .finish()
    }
}

impl NamingServiceImpl {
    pub fn new(
        service_storage: Arc<dyn ServiceStorage>,
        event_publisher: Arc<dyn EventPublisher>,
        network_service: Arc<dyn NetworkService>,
    ) -> Self {
        info!("Creating new NamingServiceImpl");
        Self {
            service_storage,
            event_publisher,
            network_service,
            service_registry: Arc::new(RwLock::new(ServiceRegistry::new())),
        }
    }

    /// 获取服务状态
    pub async fn get_service_status(&self, namespace: &str, group: &str, service_name: &str) -> Result<Option<ServiceStatus>> {
        let registry = self.service_registry.read().await;
        Ok(registry.get_service_status(namespace, group, service_name))
    }

    /// 获取所有服务名
    pub async fn get_all_service_names(&self, namespace: &str, group: Option<&str>) -> Result<Vec<String>> {
        let registry = self.service_registry.read().await;
        Ok(registry.get_all_service_names(namespace, group))
    }
}

#[async_trait]
impl NamingService for NamingServiceImpl {
    #[instrument(skip(self, instance))]
    async fn register_instance(&self, instance: &ServiceInstance) -> Result<()> {
        info!("Registering service instance: {}/{}/{}", instance.namespace, instance.group_name, instance.service_name);
        
        // 先存储到持久化存储
        self.service_storage.register_instance(instance).await?;
        
        // 再注册到内存注册表
        let mut registry = self.service_registry.write().await;
        registry.register_instance(instance.clone());
        
        info!("Successfully registered instance: {}", instance.instance_id);
        Ok(())
    }

    #[instrument(skip(self))]
    async fn deregister_instance(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str) -> Result<()> {
        info!("Deregistering service instance: {}/{}/{}/{}", namespace, group, service_name, instance_id);
        
        // 先从持久化存储删除
        self.service_storage.deregister_instance(namespace, group, service_name, instance_id).await?;
        
        // 再从内存注册表删除
        let mut registry = self.service_registry.write().await;
        registry.deregister_instance(namespace, group, service_name, instance_id);
        
        info!("Successfully deregistered instance: {}", instance_id);
        Ok(())
    }

    #[instrument(skip(self))]
    async fn get_healthy_instances(&self, namespace: &str, group: &str, service_name: &str) -> Result<Vec<ServiceInstance>> {
        info!("Getting healthy instances: {}/{}/{}", namespace, group, service_name);
        
        // 优先从内存注册表获取，提高性能
        let registry = self.service_registry.read().await;
        let instances = registry.get_healthy_instances(namespace, group, service_name);
        
        // 将引用转换为拥有的实例（如果需要）
        Ok(instances.into_iter().cloned().collect())
    }

    #[instrument(skip(self))]
    async fn get_all_instances(&self, namespace: &str, group: &str, service_name: &str) -> Result<Vec<ServiceInstance>> {
        info!("Getting all instances: {}/{}/{}", namespace, group, service_name);
        
        // 优先从内存注册表获取，提高性能
        let registry = self.service_registry.read().await;
        let instances = registry.get_all_instances(namespace, group, service_name);
        
        // 将引用转换为拥有的实例
        Ok(instances.into_iter().cloned().collect())
    }

    #[instrument(skip(self))]
    async fn get_instance(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str) -> Result<Option<ServiceInstance>> {
        info!("Getting instance: {}/{}/{}/{}", namespace, group, service_name, instance_id);
        
        // 优先从内存注册表获取，提高性能
        let registry = self.service_registry.read().await;
        let instance = registry.get_instance(namespace, group, service_name, instance_id);
        
        if let Some(instance) = instance {
            return Ok(Some(instance.clone()));
        }
        
        // 如果内存中没有，从持久化存储获取
        self.service_storage.get_instance(namespace, group, service_name, instance_id).await
    }

    #[instrument(skip(self, instance))]
    async fn update_instance(&self, instance: &ServiceInstance) -> Result<()> {
        info!("Updating service instance: {}", instance.instance_id);
        
        // 更新持久化存储
        self.service_storage.register_instance(instance).await?;
        
        // 更新内存注册表
        let mut registry = self.service_registry.write().await;
        registry.register_instance(instance.clone());
        
        info!("Successfully updated instance: {}", instance.instance_id);
        Ok(())
    }

    #[instrument(skip(self))]
    async fn get_all_service_names(&self, namespace: &str, group: Option<&str>) -> Result<Vec<String>> {
        info!("Getting all service names: namespace={}, group={:?}", namespace, group);
        
        // 优先从内存注册表获取，提高性能
        let registry = self.service_registry.read().await;
        let services = registry.get_all_service_names(namespace, group);
        
        // 如果内存中没有，从持久化存储获取
        if services.is_empty() {
            self.service_storage.get_all_services(namespace, group).await
        } else {
            Ok(services)
        }
    }

    #[instrument(skip(self, _listener))]
    async fn subscribe_service(&self, _namespace: &str, _group: &str, _service_name: &str, _listener: Box<dyn EventListener>) -> Result<()> {
        Err(NacosError::Unknown("Not implemented".to_string()))
    }

    #[instrument(skip(self))]
    async fn unsubscribe_service(&self, _namespace: &str, _group: &str, _service_name: &str, _listener_name: &str) -> Result<()> {
        Err(NacosError::Unknown("Not implemented".to_string()))
    }

    #[instrument(skip(self))]
    async fn handle_heartbeat(&self, namespace: &str, group: &str, service_name: &str, instance_id: &str) -> Result<()> {
        info!("Handling heartbeat for instance: {}/{}/{}/{}", namespace, group, service_name, instance_id);

        // Nacos 1.x/2.x client beat use `group@@service_name` as service_name
        let service_name = if let Some(index) = service_name.find("@@") {
            &service_name[index + 2..]
        } else {
            service_name
        };
        
        // 先从内存注册表查找
        let mut registry = self.service_registry.write().await;
        
        if let Some(instance) = registry.get_instance(namespace, group, service_name, instance_id) {
            let mut updated_instance = instance.clone();
            updated_instance.last_heartbeat_time = chrono::Utc::now();
            
            // 更新内存注册表
            registry.register_instance(updated_instance.clone());
            
            // 更新持久化存储
            self.service_storage.register_instance(&updated_instance).await?;
            
            info!("Updated heartbeat time for instance: {}", instance_id);
        } else {
            // 如果内存中没有，尝试从持久化存储获取
            if let Some(mut instance) = self.service_storage.get_instance(namespace, group, service_name, instance_id).await? {
                instance.last_heartbeat_time = chrono::Utc::now();
                
                // 同步到内存注册表
                registry.register_instance(instance.clone());
                
                // 更新持久化存储
                self.service_storage.register_instance(&instance).await?;
                
                info!("Updated heartbeat time for instance from storage: {}", instance_id);
            } else {
                warn!("Instance not found for heartbeat: {}/{}/{}/{}", namespace, group, service_name, instance_id);
                return Err(NacosError::NotFound(format!("Instance not found: {}", instance_id)));
            }
        }
        Ok(())
    }
}

// --- Mock Event Publisher ---
#[derive(Debug, Default)]
pub struct MockEventPublisher;

#[async_trait]
impl EventPublisher for MockEventPublisher {
    async fn publish(&self, _event: Event) -> Result<()> { Ok(()) } 
    async fn subscribe(&self, _listener: Box<dyn EventListener>) -> Result<()> { Ok(()) }
    async fn unsubscribe(&self, _listener_name: &str) -> Result<()> { Ok(()) }
}

impl NamingServiceImpl {
    /// 启动命名服务
    ///
    /// # 返回值
    /// - `Ok(())`: 服务启动成功
    /// - `Err(NacosError)`: 启动失败时的错误
    ///
    /// # 行为
    /// 初始化命名服务，加载持久化数据到内存注册表。
    pub async fn start(&self) -> Result<()> {
        info!("Starting naming service");
        
        // 从持久化存储加载所有服务到内存注册表
        let all_services = self.service_storage.get_all_services("public", None).await?;
        info!("Loaded {} services from storage", all_services.len());
        
        // 初始化内存注册表
        let mut registry = self.service_registry.write().await;
        for service_name in all_services {
            let instances = self.service_storage
                .get_all_instances("public", "DEFAULT_GROUP", &service_name)
                .await?;
            
            for instance in instances {
                registry.register_instance(instance);
            }
        }
        
        info!("Naming service started successfully");
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use nacos_core::{
        ClusterNode,
        traits::{NetworkRequest, NetworkResponse},
    };
    use nacos_storage::memory::MemoryServiceStorage;

    // --- Mock Network Service ---
    #[derive(Debug)]
    struct MockNetworkService;

    #[async_trait]
    impl NetworkService for MockNetworkService {
        async fn send_request(&self, _node: &ClusterNode, _request: &NetworkRequest) -> Result<NetworkResponse> {
            Ok(NetworkResponse::default())
        }

        async fn broadcast(&self, _request: &NetworkRequest) -> Result<Vec<NetworkResponse>> {
            Ok(vec![])
        }

        async fn connect(&self, _node: &ClusterNode) -> Result<()> {
            Ok(())
        }

        async fn disconnect(&self, _node: &ClusterNode) -> Result<()> {
            Ok(())
        }

        async fn is_connected(&self, _node: &ClusterNode) -> Result<bool> {
            Ok(true)
        }
    }

    // --- Reusable Test Suite ---
    async fn run_naming_service_test_suite(service: impl NamingService, storage: Arc<dyn ServiceStorage>) {
        let mut instance = ServiceInstance::new(
            "test_service".to_string(),
            "test_group".to_string(),
            "127.0.0.1".to_string(),
            8080,
        );
        instance.namespace = "public".to_string();

        // 1. Register a new instance
        service.register_instance(&instance).await.unwrap();

        // 2. Get the instance and verify
        let retrieved = storage.get_instance("public", "test_group", "test_service", &instance.instance_id).await.unwrap().unwrap();
        assert_eq!(retrieved, instance);

        // 3. Get all instances
        let all = service.get_all_instances("public", "test_group", "test_service").await.unwrap();
        assert_eq!(all.len(), 1);

        // 4. Get healthy instances
        let healthy = service.get_healthy_instances("public", "test_group", "test_service").await.unwrap();
        assert_eq!(healthy.len(), 1);

        // 5. Update instance health
        let mut updated_instance = instance.clone();
        updated_instance.healthy = false;
        service.update_instance(&updated_instance).await.unwrap();
        let healthy_after_update = service.get_healthy_instances("public", "test_group", "test_service").await.unwrap();
        assert_eq!(healthy_after_update.len(), 0);

        // 6. Get all service names
        let services = service.get_all_service_names("public", Some("test_group")).await.unwrap();
        assert!(services.contains(&"test_service".to_string()));

        // 7. Deregister the instance
        service.deregister_instance("public", "test_group", "test_service", &instance.instance_id).await.unwrap();
        let after_deregister = service.get_all_instances("public", "test_group", "test_service").await.unwrap();
        assert!(after_deregister.is_empty());
    }

    #[tokio::test]
    async fn test_naming_service_with_memory_storage() {
        let storage = Arc::new(MemoryServiceStorage::new());
        let publisher = Arc::new(MockEventPublisher);
        let network = Arc::new(MockNetworkService);
        let service = NamingServiceImpl::new(storage.clone(), publisher, network);
        
        run_naming_service_test_suite(service, storage).await;
    }

    #[tokio::test]
    async fn test_service_status_management() {
        let storage = Arc::new(MemoryServiceStorage::new());
        let publisher = Arc::new(MockEventPublisher);
        let network = Arc::new(MockNetworkService);
        let service = NamingServiceImpl::new(storage.clone(), publisher, network);

        // 注册多个实例
        let mut instance1 = ServiceInstance::new(
            "test_service".to_string(),
            "test_group".to_string(),
            "127.0.0.1".to_string(),
            8080,
        );
        instance1.namespace = "public".to_string();
        instance1.healthy = true;

        let mut instance2 = ServiceInstance::new(
            "test_service".to_string(),
            "test_group".to_string(),
            "127.0.0.1".to_string(),
            8081,
        );
        instance2.namespace = "public".to_string();
        instance2.healthy = false;

        service.register_instance(&instance1).await.unwrap();
        service.register_instance(&instance2).await.unwrap();

        // 测试服务状态
        let status = service.get_service_status("public", "test_group", "test_service").await.unwrap();
        assert!(status.is_some());
        let status = status.unwrap();
        assert_eq!(status.service_name, "test_service");
        assert_eq!(status.total_instance_count, 2);
        assert_eq!(status.healthy_instance_count, 1);
    }

    #[tokio::test]
    async fn test_heartbeat_functionality() {
        let storage = Arc::new(MemoryServiceStorage::new());
        let publisher = Arc::new(MockEventPublisher);
        let network = Arc::new(MockNetworkService);
        let service = NamingServiceImpl::new(storage.clone(), publisher, network);

        let mut instance = ServiceInstance::new(
            "test_service".to_string(),
            "test_group".to_string(),
            "127.0.0.1".to_string(),
            8080,
        );
        instance.namespace = "public".to_string();
        let original_heartbeat = instance.last_heartbeat_time;

        service.register_instance(&instance).await.unwrap();

        // 等待一小段时间
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        // 处理心跳
        service.handle_heartbeat("public", "test_group", "test_service", &instance.instance_id).await.unwrap();

        // 验证心跳时间已更新
        let retrieved = service.get_instance("public", "test_group", "test_service", &instance.instance_id).await.unwrap().unwrap();
        assert!(retrieved.last_heartbeat_time > original_heartbeat);
    }

    #[tokio::test]
    async fn test_multiple_services_management() {
        let storage = Arc::new(MemoryServiceStorage::new());
        let publisher = Arc::new(MockEventPublisher);
        let network = Arc::new(MockNetworkService);
        let service = NamingServiceImpl::new(storage.clone(), publisher, network);

        // 注册多个服务
        let mut instance1 = ServiceInstance::new(
            "service_a".to_string(),
            "group_1".to_string(),
            "192.168.1.1".to_string(),
            8080,
        );
        instance1.namespace = "public".to_string();
        let mut instance2 = ServiceInstance::new(
            "service_b".to_string(),
            "group_1".to_string(),
            "192.168.1.2".to_string(),
            8081,
        );
        instance2.namespace = "public".to_string();
        let mut instance3 = ServiceInstance::new(
            "service_a".to_string(),
            "group_2".to_string(),
            "192.168.1.3".to_string(),
            8082,
        );
        instance3.namespace = "public".to_string();

        service.register_instance(&instance1).await.unwrap();
        service.register_instance(&instance2).await.unwrap();
        service.register_instance(&instance3).await.unwrap();

        // 测试按命名空间和分组查询
        let services_group1 = service.get_all_service_names("public", Some("group_1")).await.unwrap();
        assert_eq!(services_group1.len(), 2);
        assert!(services_group1.contains(&"service_a".to_string()));
        assert!(services_group1.contains(&"service_b".to_string()));

        let services_group2 = service.get_all_service_names("public", Some("group_2")).await.unwrap();
        assert_eq!(services_group2.len(), 1);
        assert!(services_group2.contains(&"service_a".to_string()));

        let all_services = service.get_all_service_names("public", None).await.unwrap();
        assert_eq!(all_services.len(), 2);
    }

    #[tokio::test]
    async fn test_handle_heartbeat() {
        let storage = Arc::new(MemoryServiceStorage::new());
        let publisher = Arc::new(MockEventPublisher);
        let network = Arc::new(MockNetworkService);
        let service = NamingServiceImpl::new(storage.clone(), publisher, network);

        let mut instance = ServiceInstance::new(
            "test_service".to_string(),
            "test_group".to_string(),
            "127.0.0.1".to_string(),
            8080,
        );
        instance.namespace = "public".to_string();
        let original_heartbeat = instance.last_heartbeat_time;

        service.register_instance(&instance).await.unwrap();

        // Wait a bit to ensure the heartbeat time changes
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        service.handle_heartbeat("public", "test_group", "test_service", &instance.instance_id).await.unwrap();

        let updated_instance = storage.get_instance("public", "test_group", "test_service", &instance.instance_id).await.unwrap().unwrap();
        assert!(updated_instance.last_heartbeat_time > original_heartbeat);
    }
}